package com.zhj.kafka;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;

/**
 * 　　　　　　　   ┏┓　   ┏┓+ +
 * 　　　　　　　┏┛┻━━━┛┻┓ + +
 * 　　　　　　　┃　　　　　　　┃
 * 　　　　　　　┃　　　━　　　┃ ++ + + +
 * 　　　　　　 ████━████ ┃+
 * 　　　　　　　┃　　　　　　　┃ +
 * 　　　　　　　┃　　　┻　　　┃
 * 　　　　　　　┃　　　　　　　┃ + +
 * 　　　　　　　┗━┓　　　┏━┛
 * 　　　　　　　　　┃　　　┃
 * 　　　　　　　　　┃　　　┃ + + + +
 * 　　　　　　　　　┃　　　┃			God beast body, code no BUG
 * 　　　　　　　　　┃　　　┃ +			神兽护体,代码无BUG
 * 　　　　　　　　　┃　　　┃
 * 　　　　　　　　　┃　　　┃　　+
 * 　　　　　　　　　┃　 　　┗━━━┓ + +
 * 　　　　　　　　　┃ 　　　　　　　┣┓
 * 　　　　　　　　　┃ 　　　　　　　┏┛
 * 　　　　　　　　　┗┓┓┏━┳┓┏┛ + + + +
 * 　　　　　　　　　　┃┫┫　┃┫┫
 * 　　　　　　　　　　┗┻┛　┗┻┛+ + + +
 *
 * @author zhonghaijun
 * @title:
 * @date 2021 -04-15
 */
public class KafkaConsumerApplication {

    /**
     * Get properties properties.
     *
     * @return the properties
     */
    public static Properties getProperties(){
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "47.116.130.99:21461");
        //消费组名
        properties.put("timeout.ms",3000);

        //自动提交offset的间隔时间
        properties.put("auto.commit.interval.ms","1000");
        //是否开启自动提交
        properties.put("enable.auto.commit",false);
        properties.put("zookeeper.connect","localhost:2181");
        properties.put("group.id","test1");
        //把发送的key从字符串序列化为字节数组
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //把发送的value从字符串序列化为字节数组
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put("partition.assignment.strategy","range");
        return properties;
    }


    /**
     * The entry point of application.
     *
     * @param args the input arguments
     * @throws InterruptedException the interrupted exception
     */
    public static void main(String[] args) throws InterruptedException {
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(getProperties());
        String topicName = "test";
        //kafkaConsumer.subscribe(Arrays.asList(topicName));
        TopicPartition topicPartition = new TopicPartition(topicName,0);
        //指定偏移量进行消费
        /**
         * assign():会将需要消费的分区进行注册到SubscriptionState
         * 如果 只使用 subscribe()方法 只会注册整个分区
         */
        //kafkaConsumer.subscribe(topicPartition);
        kafkaConsumer.assign(Collections.singletonList(topicPartition));
        /*HashMap<TopicPartition, Long> patitionsMap = new HashMap<>();
        patitionsMap.put(topicPartition,0L);
        kafkaConsumer.seek(patitionsMap);*/
        kafkaConsumer.seekToBeginning(Arrays.asList(topicPartition));  //指定分区消费
        //订阅主题
        /**
         * poll() 长轮询，主要判断consumer是否还活着
         * 底层：间隔一段时间就会发送心跳到broker，如果session.timeout.ms的时间没有发送，就会被提出group，partition也会重新分配
         */
        retry:
        while (true){
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofNanos(1000));
            if (consumerRecords.count() <= 0){
                continue;
            }
            /*consumerRecords.forEach(new BiConsumer<String, ConsumerRecords<String, String>>() {
                @Override
                public void accept(String s, ConsumerRecords<String, String> stringStringConsumerRecords) {
                    System.out.println("消息："+s);
                    List<ConsumerRecord<String, String>> consumerRecords = stringStringConsumerRecords.records();
                    for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                        System.out.println("--------------------------------------");
                        try {
                            System.out.println("收到消息：offset："+consumerRecord.offset()+" key："+consumerRecord.key()+" value："+consumerRecord.value());
                        } catch (Exception e) {
                            System.out.println("接收消息异常。。。");
                        }
                        System.out.println("--------------------------------------");
                    }
                }
            });*/
            //如果有消费的信息，那么手动提交offset
            if(consumerRecords.count() > 0){
                kafkaConsumer.commitAsync();
            }
            Thread.sleep(2000);
        }
    }
}
